{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Ifon2ScEAsJO"
      },
      "source": [
        "<div class=\"devsite-table-wrapper\"><table class=\"tfo-notebook-buttons\" align=\"left\">\n",
        "<td><a target=\"_blank\" href=\"https://www.tensorflow.org/tfx/tutorials/transform/simple\">\n",
        "<img src=\"https://www.tensorflow.org/images/tf_logo_32px.png\" />View on TensorFlow.org</a></td>\n",
        "<td><a target=\"_blank\" href=\"https://colab.research.google.com/github/tensorflow/tfx/blob/master/docs/tutorials/transform/simple.ipynb\">\n",
        "<img src=\"https://www.tensorflow.org/images/colab_logo_32px.png\">Run in Google Colab</a></td>\n",
        "<td><a target=\"_blank\" href=\"https://github.com/tensorflow/tfx/blob/master/docs/tutorials/transform/simple.ipynb\">\n",
        "<img width=32px src=\"https://www.tensorflow.org/images/GitHub-Mark-32px.png\">View source on GitHub</a></td>\n",
        "</table></div>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tghWegsjhpkt"
      },
      "source": [
        "##### Copyright &copy; 2020 Google Inc."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "rSGJWC5biBiG"
      },
      "outputs": [],
      "source": [
        "#@title Licensed under the Apache License, Version 2.0 (the \"License\");\n",
        "# you may not use this file except in compliance with the License.\n",
        "# You may obtain a copy of the License at\n",
        "#\n",
        "# https://www.apache.org/licenses/LICENSE-2.0\n",
        "#\n",
        "# Unless required by applicable law or agreed to in writing, software\n",
        "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
        "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
        "# See the License for the specific language governing permissions and\n",
        "# limitations under the License."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "mPt5BHTwy_0F"
      },
      "source": [
        "# Preprocess data with TensorFlow Transform\n",
        "***The Feature Engineering Component of TensorFlow Extended (TFX)***\n",
        "\n",
        "This example colab notebook provides a very simple example of how <a target='_blank' href='https://www.tensorflow.org/tfx/transform/'>TensorFlow Transform (<code>tf.Transform</code>)</a> can be used to preprocess data using exactly the same code for both training a model and serving inferences in production.\n",
        "\n",
        "TensorFlow Transform is a library for preprocessing input data for TensorFlow, including creating features that require a full pass over the training dataset.  For example, using TensorFlow Transform you could:\n",
        "\n",
        "* Normalize an input value by using the mean and standard deviation\n",
        "* Convert strings to integers by generating a vocabulary over all of the input values\n",
        "* Convert floats to integers by assigning them to buckets, based on the observed data distribution\n",
        "\n",
        "TensorFlow has built-in support for manipulations on a single example or a batch of examples. `tf.Transform` extends these capabilities to support full passes over the entire training dataset.\n",
        "\n",
        "The output of `tf.Transform` is exported as a TensorFlow graph which you can use for both training and serving. Using the same graph for both training and serving can prevent skew, since the same transformations are applied in both stages."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6c8lD3uQm8m5"
      },
      "source": [
        "### Upgrade Pip\n",
        "\n",
        "To avoid upgrading Pip in a system when running locally, check to make sure that we're running in Colab.  Local systems can of course be upgraded separately."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "EmiQXNLZm8z-"
      },
      "outputs": [],
      "source": [
        "try:\n",
        "  import colab\n",
        "  !pip install --upgrade pip\n",
        "except:\n",
        "  pass"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hiBxgnc-m8-X"
      },
      "source": [
        "### Install TensorFlow Transform\n",
        "\n",
        "**Note: In Google Colab, because of package updates, the first time you run this cell you may need to restart the runtime (Runtime > Restart runtime ...).**"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "j2CTKbMNm9I4"
      },
      "outputs": [],
      "source": [
        "!pip install -q -U tensorflow_transform==0.24.1"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "PdIIROKiZpdy"
      },
      "source": [
        "## Did you restart the runtime?\n",
        "\n",
        "If you are using Google Colab, the first time that you run the cell above, you must restart the runtime (Runtime > Restart runtime ...). This is because of the way that Colab loads packages."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "RptgLn2RYuK3"
      },
      "source": [
        "## Imports"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "K4QXVIM7iglN"
      },
      "outputs": [],
      "source": [
        "import pprint\n",
        "import tempfile\n",
        "\n",
        "import tensorflow as tf\n",
        "import tensorflow_transform as tft\n",
        "\n",
        "import tensorflow_transform.beam as tft_beam\n",
        "from tensorflow_transform.tf_metadata import dataset_metadata\n",
        "from tensorflow_transform.tf_metadata import schema_utils"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CxOxaaOYRfl7"
      },
      "source": [
        "## Data: Create some dummy data\n",
        "We'll create some simple dummy data for our simple example:\n",
        "\n",
        "* `raw_data` is the initial raw data that we're going to preprocess\n",
        "* `raw_data_metadata` contains the schema that tells us the types of each of the columns in `raw_data`.  In this case, it's very simple."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "-R236Tkf_ON3"
      },
      "outputs": [],
      "source": [
        "raw_data = [\n",
        "      {'x': 1, 'y': 1, 's': 'hello'},\n",
        "      {'x': 2, 'y': 2, 's': 'world'},\n",
        "      {'x': 3, 'y': 3, 's': 'hello'}\n",
        "  ]\n",
        "\n",
        "raw_data_metadata = dataset_metadata.DatasetMetadata(\n",
        "    schema_utils.schema_from_feature_spec({\n",
        "        'y': tf.io.FixedLenFeature([], tf.float32),\n",
        "        'x': tf.io.FixedLenFeature([], tf.float32),\n",
        "        's': tf.io.FixedLenFeature([], tf.string),\n",
        "    }))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Zadh6MXLS3eD"
      },
      "source": [
        "## Transform: Create a preprocessing function\n",
        "The _preprocessing function_ is the most important concept of tf.Transform. A preprocessing function is where the transformation of the dataset really happens. It accepts and returns a dictionary of tensors, where a tensor means a <a target='_blank' href='https://www.tensorflow.org/versions/r1.15/api_docs/python/tf/Tensor'><code>Tensor</code></a> or <a target='_blank' href='https://www.tensorflow.org/versions/r1.15/api_docs/python/tf/SparseTensor'><code>SparseTensor</code></a>. There are two main groups of API calls that typically form the heart of a preprocessing function:\n",
        "\n",
        "1. **TensorFlow Ops:** Any function that accepts and returns tensors, which usually means TensorFlow ops. These add TensorFlow operations to the graph that transforms raw data into transformed data one feature vector at a time.  These will run for every example, during both training and serving.\n",
        "2. **Tensorflow Transform Analyzers/Mappers:** Any of the analyzers/mappers provided by tf.Transform. These also accept and return tensors, and typically contain a combination of Tensorflow ops and Beam computation, but unlike TensorFlow ops they only run in the Beam pipeline during analysis requiring a full pass over the entire training dataset. The Beam computation runs only once, during training, and typically make a full pass over the entire training dataset. They create tensor constants, which are added to your graph. For example, tft.min computes the minimum of a tensor over the training dataset while tft.scale_by_min_max first computes the min and max of a tensor over the training dataset and then scales the tensor to be within a user-specified range, [output_min, output_max]. tf.Transform provides a fixed set of such analyzers/mappers, but this will be extended in future versions.\n",
        "\n",
        "Caution: When you apply your preprocessing function to serving inferences, the constants that were created by analyzers during training do not change.  If your data has trend or seasonality components, plan accordingly.\n",
        "\n",
        "Note: The `preprocessing_fn` is not directly callable. This means that\n",
        "calling `preprocessing_fn(raw_data)` will not work. Instead, it must\n",
        "be passed to the Transform Beam API as shown in the following cells."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "H2wANNF_2dCR"
      },
      "outputs": [],
      "source": [
        "def preprocessing_fn(inputs):\n",
        "    \"\"\"Preprocess input columns into transformed columns.\"\"\"\n",
        "    x = inputs['x']\n",
        "    y = inputs['y']\n",
        "    s = inputs['s']\n",
        "    x_centered = x - tft.mean(x)\n",
        "    y_normalized = tft.scale_to_0_1(y)\n",
        "    s_integerized = tft.compute_and_apply_vocabulary(s)\n",
        "    x_centered_times_y_normalized = (x_centered * y_normalized)\n",
        "    return {\n",
        "        'x_centered': x_centered,\n",
        "        'y_normalized': y_normalized,\n",
        "        's_integerized': s_integerized,\n",
        "        'x_centered_times_y_normalized': x_centered_times_y_normalized,\n",
        "    }"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cSl9qyTCbBKR"
      },
      "source": [
        "## Putting it all together\n",
        "Now we're ready to transform our data.  We'll use Apache Beam with a direct runner, and supply three inputs:\n",
        "1. `raw_data` - The raw input data that we created above\n",
        "2. `raw_data_metadata` - The schema for the raw data\n",
        "3. `preprocessing_fn` - The function that we created to do our transformation\n",
        "\n",
        "<aside class=\"key-term\"><b>Key Term:</b> <a target='_blank' href='https://beam.apache.org/'>Apache Beam</a> uses a <a target='_blank' href='https://beam.apache.org/documentation/programming-guide/#applying-transforms'>special syntax to define and invoke transforms</a>.  For example, in this line:\n",
        "\n",
        "<code><blockquote>result = pass_this | 'name this step' >> to_this_call</blockquote></code>\n",
        "\n",
        "The method <code>to_this_call</code> is being invoked and passed the object called <code>pass_this</code>, and <a target='_blank' href='https://stackoverflow.com/questions/50519662/what-does-the-redirection-mean-in-apache-beam-python'>this operation will be referred to as <code>name this step</code> in a stack trace</a>.  The result of the call to <code>to_this_call</code> is returned in <code>result</code>.  You will often see stages of a pipeline chained together like this:\n",
        "\n",
        "<code><blockquote>result = apache_beam.Pipeline() | 'first step' >> do_this_first() | 'second step' >> do_this_last()</blockquote></code>\n",
        "\n",
        "and since that started with a new pipeline, you can continue like this:\n",
        "\n",
        "<code><blockquote>next_result = result | 'doing more stuff' >> another_function()</blockquote></code></aside>"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "mAF9w7RTZU7c"
      },
      "outputs": [],
      "source": [
        "def main():\n",
        "  # Ignore the warnings\n",
        "  with tft_beam.Context(temp_dir=tempfile.mkdtemp()):\n",
        "    transformed_dataset, transform_fn = (  # pylint: disable=unused-variable\n",
        "        (raw_data, raw_data_metadata) | tft_beam.AnalyzeAndTransformDataset(\n",
        "            preprocessing_fn))\n",
        "\n",
        "  transformed_data, transformed_metadata = transformed_dataset  # pylint: disable=unused-variable\n",
        "\n",
        "  print('\\nRaw data:\\n{}\\n'.format(pprint.pformat(raw_data)))\n",
        "  print('Transformed data:\\n{}'.format(pprint.pformat(transformed_data)))\n",
        "\n",
        "if __name__ == '__main__':\n",
        "  main()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "NO6LyTneNndy"
      },
      "source": [
        "## Is this the right answer?\n",
        "Previously, we used `tf.Transform` to do this:\n",
        "```\n",
        "x_centered = x - tft.mean(x)\n",
        "y_normalized = tft.scale_to_0_1(y)\n",
        "s_integerized = tft.compute_and_apply_vocabulary(s)\n",
        "x_centered_times_y_normalized = (x_centered * y_normalized)\n",
        "```\n",
        "####x_centered\n",
        "With input of `[1, 2, 3]` the mean of x is 2, and we subtract it from x to center our x values at 0.  So our result of `[-1.0, 0.0, 1.0]` is correct.\n",
        "####y_normalized\n",
        "We wanted to scale our y values between 0 and 1.  Our input was `[1, 2, 3]` so our result of `[0.0, 0.5, 1.0]` is correct.\n",
        "####s_integerized\n",
        "We wanted to map our strings to indexes in a vocabulary, and there were only 2 words in our vocabulary (\"hello\" and \"world\").  So with input of `[\"hello\", \"world\", \"hello\"]` our result of `[0, 1, 0]` is correct. Since \"hello\" occurs most frequently in this data, it will be the first entry in the vocabulary.\n",
        "####x_centered_times_y_normalized\n",
        "We wanted to create a new feature by crossing `x_centered` and `y_normalized` using multiplication.  Note that this multiplies the results, not the original values, and our new result of `[-0.0, 0.0, 1.0]` is correct."
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [
        "tghWegsjhpkt"
      ],
      "name": "simple.ipynb",
      "toc_visible": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}
